package com.wlyuan.index.domain.index.service.reindex;

import com.alibaba.fastjson.JSON;
import com.wlyuan.core.domain.valueobj.OrderId;
import com.wlyuan.index.api.dto.order.OrderReindexAllDTO;
import com.wlyuan.index.api.dto.order.OrderReindexDTO;
import com.wlyuan.index.domain.index.factory.OrderIndexFactory;
import com.wlyuan.index.domain.source.Source;
import com.wlyuan.index.domain.source.factory.SourceFactory;
import com.wlyuan.index.domain.source.repository.SourceRepository;
import com.wlyuan.index.remote.facade.OrderFacade;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import lombok.var;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;

import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 * @author yuanjie
 */
@Service
@Slf4j
@RequiredArgsConstructor
public class OrderReindexService {
    private final SourceFactory sourceFactory;
    private final SourceRepository sourceRepository;
    private final OrderIndexFactory indexFactory;
    private final OrderFacade orderFacade;
    private static final ExecutorService EXECUTOR = new ThreadPoolExecutor(4,
            4, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(),
            new CustomizableThreadFactory("order-reindex-pool"));

    public void reindex(OrderId orderId) {
        try {
            var order = orderFacade.getOrderById(orderId);
            if (null == order) {
                logger.error("订单不存在: {}", orderId);
                return;
            }

            sourceFactory.create(order);
            // 创建索引
            indexFactory.create(orderId);
        } catch (Exception e) {
            logger.error("重建订单索引异常: {}", orderId, e);
        }
    }

    public void reindex(OrderReindexDTO reindex) {
        logger.info("重建订单索引: {}", JSON.toJSON(reindex));
        CompletableFuture.runAsync(() -> {
            var page = 1;
            var size = 1000;
            var totalSize = new AtomicLong(0);
            var completedSize = new AtomicLong(0L);
            var orderIds = orderFacade.listIds(reindex, page++, size);
            while (CollectionUtils.isNotEmpty(orderIds)) {
                logger.info("创建重建订单索引任务: {} {} {}", reindex, page - 1, size);
                this.reindexAsync(orderIds, completedSize);
                totalSize.addAndGet(orderIds.size());
                orderIds = orderFacade.listIds(reindex, page++, size);
            }
            logger.info("重建订单索引任务提交完成: {} {} {}", reindex, page - 2, totalSize.get());
        });
    }


    private void reindexAsync(List<OrderId> orderIds, AtomicLong completedSize) {
        CompletableFuture.runAsync(() -> {
            if (CollectionUtils.isEmpty(orderIds)) {
                return;
            }
            StopWatch watch = new StopWatch();
            watch.start();
            orderIds.forEach(orderId -> {
                try {
                    this.reindex(orderId);
                } catch (Exception e) {
                    logger.error("重建订单索引异常: {}", orderId, e);
                }
            });
            watch.stop();
            completedSize.addAndGet(orderIds.size());
            logger.info("重建订单索引任务完成, 耗时: {}ms {} {}", watch.getTotalTimeMillis(),
                    orderIds.size(), completedSize.get());
        }, EXECUTOR);
    }

    public void reindexAsync(List<OrderId> orderIds) {
        var completedSize = new AtomicLong(0L);
        this.reindexAsync(orderIds, completedSize);
    }

    public void reindexAll(List<OrderId> orderIds) {
        var orders = orderFacade.getOrderByIds(orderIds);
        var sources = orders.stream()
                .map(Source::new)
                .collect(Collectors.toList());
        sourceRepository.saveAll(sources);
        indexFactory.create(orderIds);
    }

    public void reindex(OrderReindexAllDTO reindexAll) {
        OrderReindexDTO reindex = new OrderReindexDTO();
        BeanUtils.copyProperties(reindexAll, reindex);
        this.reindex(reindex);
    }
}
